#pragma once

#include <condition_variable>
#include <functional>
#include <thread>
#include <memory>
#include <atomic>
#include "buffer.hpp"

namespace bitlog
{
    using Functor = std::function<void(Buffer &)>;
    class AsyncLooper
    {
        private:
            std::atomic<bool> _stop;
            Buffer _pro_buf; // 生产缓冲区
            Buffer _con_buf; // 消费缓冲区
            std::mutex _mutex;
            std::condition_variable _cond_pro;
            std::condition_variable _cond_con;
            std::thread _thread; // 异步器对应工作线程
            Functor _callback; // 处理数据的回调函数

        public:
            using ptr = std::shared_ptr<AsyncLooper>;

            AsyncLooper(const Functor &callback)
                : _stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)), _callback(callback)
            { }

            ~AsyncLooper()
            {
                stop();
            }

            void stop()
            {
                _stop = true;
                _cond_con.notify_all(); // 唤醒所有工作线程
                _thread.join();
            }

            void push(const char *data, size_t len)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 条件变量控制, 若缓冲区剩余空间大于数据长度，则可以添加数据
                _cond_pro.wait(lock, [&](){ return _pro_buf.writeAbleSize() >= len; });
                _pro_buf.push(data, len);
                // 唤醒消费者对缓冲区中的数据进行处理
                _cond_con.notify_one();
            }
        
        private:
            // 线程入口函数--对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
            void threadEntry() 
            {
                while(1)
                {
                    // 判断生产缓冲区有没有数据，有则交换，没有则阻塞
                    {
                        std::unique_lock<std::mutex> lock(_mutex);
                        _cond_con.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });
                        // 退出标志被设置，且生产缓冲区已无数据，这时候再退出
                        if(_stop && _pro_buf.empty())
                        {
                            break;
                        }
                        _con_buf.swap(_pro_buf);
                        // 唤醒生产者
                        _cond_pro.notify_all();
                    }

                    // 被唤醒后，对消费缓冲区进行数据处理
                    _callback(_con_buf);

                    // 初始化消费缓冲区
                    _con_buf.reset();
                }
            }
    };
}